测试平台开发（Django 与 FastAPI 与 Vue）

# app/api/cases.py —— 测试用例路由
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.deps import get_db, require_roles
from app.schemas import CaseCreate, CaseOut

router = APIRouter(prefix="/api/v1/cases", tags=["用例管理"])

@router.post("", response_model=CaseOut, summary="创建测试用例")
async def create_case(
    payload: CaseCreate,
    db: AsyncSession = Depends(get_db),
    _user = Depends(require_roles("qa", "admin")),  # 角色鉴权依赖注入
):
    """新建用例并写入版本快照，tags 自动去重。"""
    case = await CaseService(db).create(payload, creator=_user.id)
    if not case:
        raise HTTPException(409, "同项目下用例编号已存在")
    return case

# platform_admin/cases/models.py —— Django 模型，与 FastAPI 共享同一库
from django.db import models
from django.contrib.auth.models import User

class TestCase(models.Model):
    """测试用例：与 FastAPI 侧 CaseOut 字段一一对应。"""
    title = models.CharField("用例标题", max_length=200, db_index=True)
    project = models.ForeignKey("Project", on_delete=models.CASCADE, related_name="cases")
    tags = models.JSONField("标签", default=list)              # 5.x JSONField 原生支持
    version = models.PositiveIntegerField("版本号", default=1)
    creator = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        indexes = [models.Index(fields=["project", "version"])]
        verbose_name = "测试用例"

# admin.py —— 一行注册即可获得完整 CRUD + 过滤 + 搜索 + 导出
from django.contrib import admin
@admin.register(TestCase)
class TestCaseAdmin(admin.ModelAdmin):
    list_display = ("id", "title", "project", "version", "creator", "created_at")
    list_filter = ("project", "tags", "created_at")
    search_fields = ("title",)

<!-- src/views/cases/CaseList.vue —— 用例管理列表页 -->
<script setup lang="ts">
import { ref, onMounted } from 'vue'
import { message } from 'ant-design-vue'
import type { TableColumns } from 'ant-design-vue'
import { api } from '@/api'              // Axios 实例，自动携带 JWT

interface CaseItem { id: number; title: string; tags: string[]; version: number }

const columns: TableColumns<CaseItem> = [
  { title: 'ID', dataIndex: 'id', width: 80 },
  { title: '用例标题', dataIndex: 'title', ellipsis: true },
  { title: '标签', dataIndex: 'tags', key: 'tags' },
  { title: '版本', dataIndex: 'version', width: 80 },
]
const data = ref<CaseItem[]>([])
const loading = ref(false)

async function fetchData() {             // 拉取用例列表
  loading.value = true
  try {
    const res = await api.get('/api/v1/cases')
    data.value = res.data.items
  } catch (e) {
    message.error('用例加载失败，请检查网络或权限')
  } finally {
    loading.value = false
  }
}
onMounted(fetchData)
</script>

<template>
  <a-table :columns="columns" :data-source="data" :loading="loading" row-key="id">
    <template #bodyCell="{ column, record }">
      <template v-if="column.key === 'tags'">
        <a-tag v-for="t in record.tags" :key="t" color="blue">{{ t }}</a-tag>
      </template>
    </template>
  </a-table>
</template>

# app/scheduler/k8s_job.py —— 通过 K8s API 调度测试任务为 Job
from kubernetes import client
from kubernetes.client import V1Job, V1JobSpec, V1ObjectMeta, V1PodSpec, V1Container

def build_test_job(task_id: str, spec: dict) -> V1Job:
    """构造 K8s Job：执行完即销毁，资源配额来自 spec.resource。"""
    container = V1Container(
        name="runner",
        image=f"registry.internal/test-engines/{spec['engine']}:2026.08",  # 镜像按引擎区分
        command=["sh", "-c", spec["entry"]],                                # 入口命令
        env=[{"name": k, "value": v} for k, v in spec["env"].items()],     # 环境变量注入
        resources={"requests": {"cpu": spec["resource"]["cpu"],
                                 "memory": spec["resource"]["memory"]}},
    )
    return V1Job(
        api_version="batch/v1", kind="Job",
        metadata=V1ObjectMeta(name=f"test-{task_id}", labels={"task-id": task_id}),
        spec=V1JobSpec(
            backoff_limit=0,           # 失败不重试，由平台决策
            ttl_seconds_after_finished=3600,  # 1 小时后自动清理 Pod
            template=V1PodSpec(restart_policy="Never", containers=[container]),
        ),
    )

# app/api/ws.py —— FastAPI WebSocket 实时推送
from fastapi import APIRouter, WebSocket
from redis.asyncio import Redis

router = APIRouter()

@router.websocket("/ws/tasks/{task_id}")
async def task_ws(ws: WebSocket, task_id: str):
    """订阅 Redis Pub/Sub，把 runner 的事件流推给前端。"""
    await ws.accept()
    redis = Redis.from_url("redis://redis:6379/1")
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"task:{task_id}:events")
    try:
        async for msg in pubsub.listen():
            if msg["type"] == "message":
                await ws.send_text(msg["data"].decode())  # 透传日志/进度
    finally:
        await pubsub.unsubscribe(f"task:{task_id}:events")
        await ws.close()